package com.arancia.codinglife.rocketmq.server;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class RocketMQServer {
//	@Value("${apache.rocketmq.consumer.PushConsumer}")
	private String consumerGroup = "PushConsumer";

//	@Value("${apache.rocketmq.namesrvAddr}")
	private String namesrvAddr = "localhost:9876";

	@PostConstruct
	public void defaultMQPushConsumer() {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
		consumer.setNamesrvAddr(namesrvAddr);
		try {
			consumer.subscribe("SELF_TEST_TOPIC", "push");
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
				try {
					for (MessageExt messageExt : list) {
						System.out.println("messageExt: " + messageExt); // 输出消息内容

						String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

						System.out.println("消费响应: msgId" + messageExt.getMsgId() + ", msgBody : " + messageBody);
					}
				} catch (Exception e) {
					e.printStackTrace();
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			});
			consumer.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}
